package com.example.springboot.dataSync;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.example.springboot.dataSync.service.DataSyncService;
import com.example.springboot.entity.Post;
import com.example.springboot.entity.vo.PostVo;
import com.example.springboot.service.ESService;
import com.example.springboot.service.UserService;
import com.example.springboot.utils.CommonUtil;
import com.example.springboot.utils.RedisUtil;
import org.apache.commons.lang.StringEscapeUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class DataProcessor {

    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private UserService userService;
    @Autowired
    private ESService esService;
    @Autowired
    private DataSyncService dataSyncService;
    //监听的数据库名称
    private final String databaseName = "ruoyi-vue";
    public void syncData(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                try {
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    if (databaseName.equals(entry.getHeader().getSchemaName()) && "blog_posts".equals(entry.getHeader().getTableName())) {
                        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                            if (eventType == CanalEntry.EventType.DELETE) {
                                //处理删除数据rowData.getBeforeColumnsList()
                                String id = rowData.getBeforeColumnsList().get(0).getValue();
                                syncDataToElasticSearch(id);
                                syncDataToRedis(id);
                            } else if (eventType == CanalEntry.EventType.INSERT) {
                                //处理新增数据rowData.getAfterColumnsList()
                                Post afterPost = createPostFromColumns(rowData.getAfterColumnsList());
                                PostVo  postVo=afterPost.toVo();
                                postVo.setNickname(userService.getUserById(afterPost.getAuthorId()).getNickname());
                                syncDataToElasticSearch(postVo);
                                syncDataToRedis(postVo);
                            } else {
                                //处理数据修改之后rowData.getAfterColumnsList()
                                Map<String, Object> map=new HashMap<>();
                                map.put("id", rowData.getAfterColumnsList().get(0).getValue());
                                for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
                                    if(column.getUpdated()){
                                        map.put(column.getName(), StringEscapeUtils.escapeJava(column.getValue()));
                                    }
                                }
                                syncDataToElasticSearch(map);
                                syncDataToRedis(map);
                            }
                        }
                    }
                    if (databaseName.equals(entry.getHeader().getSchemaName()) && "sys_users".equals(entry.getHeader().getTableName())) {
                        dataSyncService.processorUserSyncToElasticSearchAndRedis(rowChange,eventType);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void syncDataToRedis(PostVo postVo) throws IllegalAccessException {
        //插入
        String key = "blog_post:post:" + postVo.getId();
        Map<String, Object> map = CommonUtil.postVoToMap(postVo);
        redisUtil.hmset(key, map);
        redisUtil.expire(key, 60 * 60 * 24);
    }
    public void syncDataToRedis(Map<String, Object> map) {
        //修改
        String key = "blog_post:post:" + map.get("id");
        redisUtil.hmset(key, map);
    }
    public void syncDataToRedis(String id) {
        //删除
        String key = "blog_post:post:" + id;
        redisUtil.del(key);
    }
    public void syncDataToElasticSearch(PostVo postVo) throws IOException {
        esService.insert(postVo);
    }
    public void syncDataToElasticSearch(Map<String, Object> map) throws IOException {
        esService.update(map);
    }
    public void syncDataToElasticSearch(String id) throws IOException {
        esService.delete(id);
    }
    private Post createPostFromColumns(List<CanalEntry.Column> columns) {
        Post post = new Post();
        for (CanalEntry.Column column : columns) {
            switch (column.getName()) {
                case "id":
                    post.setId(Integer.valueOf(column.getValue()));
                    break;
                case "content":
                    post.setContent(column.getValue());
                    break;
                case "author_id":
                    post.setAuthorId(Integer.valueOf(column.getValue()));
                    break;
                case "category_id":
                    post.setCategoryId(Integer.valueOf(column.getValue()));
                    break;
                case "tags":
                    post.setTags(column.getValue());
                    break;
                case "updated_time":
                    post.setUpdatedTime(Timestamp.valueOf(column.getValue()));
                    break;
                case "published_time":
                    post.setPublishedTime(Timestamp.valueOf(column.getValue()));
                    break;
                case "status":
                    post.setStatus(Integer.valueOf(column.getValue()));
                    break;
                case "views":
                    post.setViews(Integer.valueOf(column.getValue()));
                    break;
                case "likes":
                    post.setLikes(Integer.valueOf(column.getValue()));
                    break;
                case "comments_count":
                    post.setCommentsCount(Integer.valueOf(column.getValue()));
                    break;
                case "collect_count":
                    post.setCollectCount(Integer.valueOf(column.getValue()));
                    break;
                case "article_type":
                    post.setArticleType(Integer.valueOf(column.getValue()));
                    break;
                case "cover":
                    post.setCover(column.getValue());
                    break;
                case "title":
                    post.setTitle(column.getValue());
                    break;
                case "summary":
                    post.setSummary(column.getValue());
                    break;
                default:
                    break;
            }
        }
        return post;
    }
}
